/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package org.apache.rocketmq.store.ha.autoswitch;

import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.SocketChannel;
import java.util.List;
import org.apache.rocketmq.common.ServiceThread;
import org.apache.rocketmq.common.constant.LoggerName;
import org.apache.rocketmq.common.utils.NetworkUtil;
import org.apache.rocketmq.logging.org.slf4j.Logger;
import org.apache.rocketmq.logging.org.slf4j.LoggerFactory;
import org.apache.rocketmq.remoting.netty.NettySystemConfig;
import org.apache.rocketmq.remoting.protocol.EpochEntry;
import org.apache.rocketmq.store.SelectMappedBufferResult;
import org.apache.rocketmq.store.config.MessageStoreConfig;
import org.apache.rocketmq.store.ha.FlowMonitor;
import org.apache.rocketmq.store.ha.HAConnection;
import org.apache.rocketmq.store.ha.HAConnectionState;
import org.apache.rocketmq.store.ha.io.AbstractHAReader;
import org.apache.rocketmq.store.ha.io.HAWriter;

public class AutoSwitchHAConnection implements HAConnection {

    /**
     * Handshake data protocol in syncing msg from master. Format:
     * <pre>
     * ┌─────────────────┬───────────────┬───────────┬───────────┬────────────────────────────────────┐
     * │  current state  │   body size   │   offset  │   epoch   │   EpochEntrySize * EpochEntryNums  │
     * │     (4bytes)    │   (4bytes)    │  (8bytes) │  (4bytes) │      (12bytes * EpochEntryNums)    │
     * ├─────────────────┴───────────────┴───────────┴───────────┼────────────────────────────────────┤
     * │                       Header                            │             Body                   │
     * │                                                         │                                    │
     * </pre>
     * Handshake Header protocol Format:
     * current state + body size + offset + epoch
     */
    public static final int HANDSHAKE_HEADER_SIZE = 4 + 4 + 8 + 4; // todo master 给 slave的握手请求头部 20字节

    /**
     * Transfer data protocol in syncing msg from master. Format:
     * <pre>
     * ┌─────────────────┬───────────────┬───────────┬───────────┬─────────────────────┬──────────────────┬──────────────────┐
     * │  current state  │   body size   │   offset  │   epoch   │   epochStartOffset  │   confirmOffset  │    log data      │
     * │     (4bytes)    │   (4bytes)    │  (8bytes) │  (4bytes) │      (8bytes)       │      (8bytes)    │   (data size)    │
     * ├─────────────────┴───────────────┴───────────┴───────────┴─────────────────────┴──────────────────┼──────────────────┤
     * │                                               Header                                             │       Body       │
     * │                                                                                                  │                  │
     * </pre>
     * Transfer Header protocol Format:
     * current state + body size + offset + epoch  + epochStartOffset + additionalInfo(confirmOffset)
     */
    public static final int TRANSFER_HEADER_SIZE = HANDSHAKE_HEADER_SIZE + 8 + 8;  // todo master 给slave的数据传输请求头部36字节 + body
    public static final int EPOCH_ENTRY_SIZE = 12;
    private static final Logger LOGGER = LoggerFactory.getLogger(LoggerName.STORE_LOGGER_NAME);
    private final AutoSwitchHAService haService;
    private final SocketChannel socketChannel;
    private final String clientAddress;
    private final EpochFileCache epochCache;
    private final AbstractWriteSocketService writeSocketService;
    private final ReadSocketService readSocketService;
    private final FlowMonitor flowMonitor;

    private volatile HAConnectionState currentState = HAConnectionState.HANDSHAKE;
    private volatile long slaveRequestOffset = -1;  // 握手请求master 和 slave对齐的偏移量(可以理解为他是截断偏移量)
    private volatile long slaveAckOffset = -1;   // slave最新的偏移量
    /**
     * Whether the slave have already sent a handshake message
     */
    private volatile boolean isSlaveSendHandshake = false;
    private volatile int currentTransferEpoch = -1;
    private volatile long currentTransferEpochEndOffset = 0;
    private volatile boolean isSyncFromLastFile = false;
    private volatile boolean isAsyncLearner = false;
    private volatile long slaveId = -1;

    /**
     * Last endOffset when master transfer data to slave
     */
    private volatile long lastMasterMaxOffset = -1;
    /**
     * Last time ms when transfer data to slave.
     */
    private volatile long lastTransferTimeMs = 0;

    public AutoSwitchHAConnection(AutoSwitchHAService haService, SocketChannel socketChannel,
        EpochFileCache epochCache) throws IOException {
        this.haService = haService;
        this.socketChannel = socketChannel; // socket
        this.epochCache = epochCache;
        // 客户端地址
        this.clientAddress = this.socketChannel.socket().getRemoteSocketAddress().toString();
        this.socketChannel.configureBlocking(false);
        this.socketChannel.socket().setSoLinger(false, -1);
        this.socketChannel.socket().setTcpNoDelay(true);
        if (NettySystemConfig.socketSndbufSize > 0) {
            this.socketChannel.socket().setReceiveBufferSize(NettySystemConfig.socketSndbufSize);
        }
        if (NettySystemConfig.socketRcvbufSize > 0) {
            this.socketChannel.socket().setSendBufferSize(NettySystemConfig.socketRcvbufSize);
        }
        // todo 写操作线程
        this.writeSocketService = new WriteSocketService(this.socketChannel);

        // todo 读操作线程
        this.readSocketService = new ReadSocketService(this.socketChannel);
        this.haService.getConnectionCount().incrementAndGet();
        this.flowMonitor = new FlowMonitor(haService.getDefaultMessageStore().getMessageStoreConfig());
    }

    @Override
    public void start() {
        changeCurrentState(HAConnectionState.HANDSHAKE);
        this.flowMonitor.start();
        this.readSocketService.start();
        this.writeSocketService.start();
    }

    @Override
    public void shutdown() {
        changeCurrentState(HAConnectionState.SHUTDOWN);
        this.flowMonitor.shutdown(true);
        this.writeSocketService.shutdown(true);
        this.readSocketService.shutdown(true);
        this.close();
    }

    @Override
    public void close() {
        if (this.socketChannel != null) {
            try {
                this.socketChannel.close();
            } catch (final IOException e) {
                LOGGER.error("", e);
            }
        }
    }

    public void changeCurrentState(HAConnectionState connectionState) {
        LOGGER.info("change state to {}", connectionState);
        this.currentState = connectionState;
    }

    public long getSlaveId() {
        return slaveId;
    }

    @Override
    public HAConnectionState getCurrentState() {
        return currentState;
    }

    @Override
    public SocketChannel getSocketChannel() {
        return socketChannel;
    }

    @Override
    public String getClientAddress() {
        return clientAddress;
    }

    @Override
    public long getSlaveAckOffset() {
        return slaveAckOffset;
    }

    @Override
    public long getTransferredByteInSecond() {
        return flowMonitor.getTransferredByteInSecond();
    }

    @Override
    public long getTransferFromWhere() {
        return this.writeSocketService.getNextTransferFromWhere();
    }

    private void changeTransferEpochToNext(final EpochEntry entry) {
        this.currentTransferEpoch = entry.getEpoch();
        this.currentTransferEpochEndOffset = entry.getEndOffset();
        if (entry.getEpoch() == this.epochCache.lastEpoch()) {
            // Use -1 to stand for Long.max
            this.currentTransferEpochEndOffset = -1;
        }
    }

    public boolean isAsyncLearner() {
        return isAsyncLearner;
    }

    public boolean isSyncFromLastFile() {
        return isSyncFromLastFile;
    }

    private synchronized void updateLastTransferInfo() {
        this.lastMasterMaxOffset = this.haService.getDefaultMessageStore().getMaxPhyOffset();
        this.lastTransferTimeMs = System.currentTimeMillis();
    }

    private synchronized void maybeExpandInSyncStateSet(long slaveMaxOffset) {
        /**
         * isAsyncLearner 只有slave才仅同步副本，learner不需要
         * lastMasterMaxOffset ： 此字段是master同步slave时 更新的，他的值本质是master同步时的最新偏移量
         * 因此slaveMaxOffset >= this.lastMasterMaxOffset逻辑的判断含义为
         * 如果lastMasterMaxOffset 大于 slaveMaxOffset
         * 说明，master已经向slave发送了最新数据，但slaveMaxOffset(slave目前最新的偏移量)确比lastMasterMaxOffset还小
         * 很可能说明，这个slave的请求是比master复制请求要早，那么这个请求可以不用做同步副本的判断，等待下一个slave的请求即可
         * */
        if (!this.isAsyncLearner && slaveMaxOffset >= this.lastMasterMaxOffset) {
            // todo 获取master同步slave的最新时间
            long caughtUpTimeMs = this.haService.getDefaultMessageStore().getMaxPhyOffset() == slaveMaxOffset ? System.currentTimeMillis() : this.lastTransferTimeMs;

            // todo 更新最新复制时间
            this.haService.updateConnectionLastCaughtUpTime(this.slaveId, caughtUpTimeMs);

            // todo 更新同步副本集合
            this.haService.maybeExpandInSyncStateSet(this.slaveId, slaveMaxOffset);
        }
    }

    /**
     * socket读线程，复制读取slave复制请求及相关处理
     * */
    class ReadSocketService extends ServiceThread {
        private static final int READ_MAX_BUFFER_SIZE = 1024 * 1024;
        private final Selector selector;
        private final SocketChannel socketChannel;
        private final ByteBuffer byteBufferRead = ByteBuffer.allocate(READ_MAX_BUFFER_SIZE);
        private final AbstractHAReader haReader;
        private int processPosition = 0;
        private volatile long lastReadTimestamp = System.currentTimeMillis();

        public ReadSocketService(final SocketChannel socketChannel) throws IOException {
            this.selector = NetworkUtil.openSelector();  // 多路复用器
            this.socketChannel = socketChannel;
            this.socketChannel.register(this.selector, SelectionKey.OP_READ);
            this.setDaemon(true);

            // todo 事件读取器
            haReader = new HAServerReader();

            // 注册相关回调函数
            haReader.registerHook(readSize -> {
                if (readSize > 0) {
                    ReadSocketService.this.lastReadTimestamp =
                        haService.getDefaultMessageStore().getSystemClock().now();
                }
            });
        }

        @Override
        public void run() {
            LOGGER.info(this.getServiceName() + " service started");

            while (!this.isStopped()) {
                try {
                    this.selector.select(1000);

                    /**
                     * 读取socket事件
                     * */
                    boolean ok = this.haReader.read(this.socketChannel, this.byteBufferRead);
                    if (!ok) {
                        AutoSwitchHAConnection.LOGGER.error("processReadEvent error");
                        break;
                    }

                    long interval = haService.getDefaultMessageStore().getSystemClock().now() - this.lastReadTimestamp;
                    if (interval > haService.getDefaultMessageStore().getMessageStoreConfig().getHaHousekeepingInterval()) {
                        LOGGER.warn("ha housekeeping, found this connection[" + clientAddress + "] expired, " + interval);
                        break;
                    }
                } catch (Exception e) {
                    AutoSwitchHAConnection.LOGGER.error(this.getServiceName() + " service has exception.", e);
                    break;
                }
            }

            this.makeStop();

            changeCurrentState(HAConnectionState.SHUTDOWN);

            writeSocketService.makeStop();

            haService.removeConnection(AutoSwitchHAConnection.this);

            haService.getConnectionCount().decrementAndGet();

            SelectionKey sk = this.socketChannel.keyFor(this.selector);
            if (sk != null) {
                sk.cancel();
            }

            try {
                this.selector.close();
                this.socketChannel.close();
            } catch (IOException e) {
                AutoSwitchHAConnection.LOGGER.error("", e);
            }

            flowMonitor.shutdown(true);

            AutoSwitchHAConnection.LOGGER.info(this.getServiceName() + " service end");
        }

        @Override
        public String getServiceName() {
            if (haService.getDefaultMessageStore().getBrokerConfig().isInBrokerContainer()) {
                return haService.getDefaultMessageStore().getBrokerIdentity().getIdentifier() + ReadSocketService.class.getSimpleName();
            }
            return ReadSocketService.class.getSimpleName();
        }

        class HAServerReader extends AbstractHAReader {
            @Override
            protected boolean processReadResult(ByteBuffer byteBufferRead) {
                while (true) {
                    boolean processSuccess = true;
                    int readSocketPos = byteBufferRead.position();
                    // 读取的字节数
                    int diff = byteBufferRead.position() - ReadSocketService.this.processPosition;

                    // slave的复制请求，最少都是12个字节，这里进行一下判断
                    if (diff >= AutoSwitchHAClient.MIN_HEADER_SIZE) {
                        int readPosition = ReadSocketService.this.processPosition;

                        /**
                         * 前4个字段为从节点的复制状态
                         * 注意：在AutoSwitchHA下，主从复制连接是有一个状态过程的
                         * */
                        HAConnectionState slaveState = HAConnectionState.values()[byteBufferRead.getInt(readPosition)];

                        switch (slaveState) {
                            /**
                             * 握手状态，则说明连接是刚创建
                             * 这是slave发起连接后的第一个请求，用于向master告知一些必须的信息
                             * */
                            case HANDSHAKE:
                                // SlaveBrokerId
                                // todo 从节点的节点id
                                Long slaveBrokerId = byteBufferRead.getLong(readPosition + AutoSwitchHAClient.HANDSHAKE_HEADER_SIZE - 8);

                                // 记录下来
                                AutoSwitchHAConnection.this.slaveId = slaveBrokerId;
                                // Flag(isSyncFromLastFile)
                                short syncFromLastFileFlag = byteBufferRead.getShort(readPosition + AutoSwitchHAClient.HANDSHAKE_HEADER_SIZE - 12);
                                if (syncFromLastFileFlag == 1) {
                                    AutoSwitchHAConnection.this.isSyncFromLastFile = true;
                                }
                                // Flag(isAsyncLearner role)
                                // todo 2字节的节点标识，如果是1表示这个节点是learner
                                short isAsyncLearner = byteBufferRead.getShort(readPosition + AutoSwitchHAClient.HANDSHAKE_HEADER_SIZE - 10);
                                if (isAsyncLearner == 1) {
                                    AutoSwitchHAConnection.this.isAsyncLearner = true;
                                }

                                isSlaveSendHandshake = true;  // todo slave是否发送了握手请求
                                byteBufferRead.position(readSocketPos);

                                // todo 记录最新position
                                ReadSocketService.this.processPosition += AutoSwitchHAClient.HANDSHAKE_HEADER_SIZE;
                                LOGGER.info("Receive slave handshake, slaveBrokerId:{}, isSyncFromLastFile:{}, isAsyncLearner:{}",
                                    AutoSwitchHAConnection.this.slaveId, AutoSwitchHAConnection.this.isSyncFromLastFile, AutoSwitchHAConnection.this.isAsyncLearner);
                                break;
                            case TRANSFER:
                                /**
                                 * 传输状态，主从进入数据复制阶段。
                                 * 注意：slave只负责上报自己最新的偏移量
                                 * */

                                // slave目前最新偏移量
                                long slaveMaxOffset = byteBufferRead.getLong(readPosition + 4);
                                ReadSocketService.this.processPosition += AutoSwitchHAClient.TRANSFER_HEADER_SIZE;

                                // 记录slave最新同步偏移量
                                AutoSwitchHAConnection.this.slaveAckOffset = slaveMaxOffset;
                                if (slaveRequestOffset < 0) {
                                    slaveRequestOffset = slaveMaxOffset;
                                }
                                byteBufferRead.position(readSocketPos);
                                /**
                                 * todo 核心，维护同步副本集合
                                 * */
                                maybeExpandInSyncStateSet(slaveMaxOffset);
                                AutoSwitchHAConnection.this.haService.updateConfirmOffsetWhenSlaveAck(AutoSwitchHAConnection.this.slaveId);
                                AutoSwitchHAConnection.this.haService.notifyTransferSome(AutoSwitchHAConnection.this.slaveAckOffset);
                                break;
                            default:
                                LOGGER.error("Current state illegal {}", currentState);
                                return false;
                        }

                        if (!slaveState.equals(currentState)) {
                            LOGGER.warn("Master change state from {} to {}", currentState, slaveState);
                            changeCurrentState(slaveState);
                        }
                        if (processSuccess) {
                            continue;
                        }
                    }

                    /**
                     * byteBufferRead 已经被写满
                     * */
                    if (!byteBufferRead.hasRemaining()) {
                        byteBufferRead.position(ReadSocketService.this.processPosition);
                        byteBufferRead.compact(); // 重置byteBufferRead
                        ReadSocketService.this.processPosition = 0; // processPosition 归0
                    }
                    break;
                }

                return true;
            }
        }
    }

    /**
     * master 主从写服务
     * */
    class WriteSocketService extends AbstractWriteSocketService {
        private SelectMappedBufferResult selectMappedBufferResult;

        public WriteSocketService(final SocketChannel socketChannel) throws IOException {
            super(socketChannel);
        }

        @Override
        protected int getNextTransferDataSize() {
            SelectMappedBufferResult selectResult = haService.getDefaultMessageStore().getCommitLogData(this.nextTransferFromWhere);
            if (selectResult == null || selectResult.getSize() <= 0) {
                return 0;
            }
            this.selectMappedBufferResult = selectResult;
            return selectResult.getSize();
        }

        @Override
        protected void releaseData() {
            this.selectMappedBufferResult.release();
            this.selectMappedBufferResult = null;
        }

        @Override
        protected boolean transferData(int maxTransferSize) throws Exception {

            if (null != this.selectMappedBufferResult && maxTransferSize >= 0) {
                this.selectMappedBufferResult.getByteBuffer().limit(maxTransferSize);
            }

            // Write Header
            boolean result = haWriter.write(this.socketChannel, this.byteBufferHeader);

            if (!result) {
                return false;
            }

            if (null == this.selectMappedBufferResult) {
                return true;
            }

            // Write Body
            result = haWriter.write(this.socketChannel, this.selectMappedBufferResult.getByteBuffer());

            if (result) {
                releaseData();
            }
            return result;
        }

        @Override
        protected void onStop() {
            if (this.selectMappedBufferResult != null) {
                this.selectMappedBufferResult.release();
            }
        }

        @Override
        public String getServiceName() {
            if (haService.getDefaultMessageStore().getBrokerConfig().isInBrokerContainer()) {
                return haService.getDefaultMessageStore().getBrokerIdentity().getIdentifier() + WriteSocketService.class.getSimpleName();
            }
            return WriteSocketService.class.getSimpleName();
        }
    }

    abstract class AbstractWriteSocketService extends ServiceThread {
        protected final Selector selector;
        protected final SocketChannel socketChannel;
        protected final HAWriter haWriter;

        protected final ByteBuffer byteBufferHeader = ByteBuffer.allocate(TRANSFER_HEADER_SIZE);
        // Store master epochFileCache: (Epoch + startOffset) * 1000
        private final ByteBuffer handShakeBuffer = ByteBuffer.allocate(EPOCH_ENTRY_SIZE * 1000);
        protected long nextTransferFromWhere = -1;
        protected boolean lastWriteOver = true;
        protected long lastWriteTimestamp = System.currentTimeMillis();
        protected long lastPrintTimestamp = System.currentTimeMillis();
        protected long transferOffset = 0;

        public AbstractWriteSocketService(final SocketChannel socketChannel) throws IOException {
            this.selector = NetworkUtil.openSelector();
            this.socketChannel = socketChannel;
            this.socketChannel.register(this.selector, SelectionKey.OP_WRITE);
            this.setDaemon(true);
            haWriter = new HAWriter();
            haWriter.registerHook(writeSize -> {
                flowMonitor.addByteCountTransferred(writeSize);
                if (writeSize > 0) {
                    AbstractWriteSocketService.this.lastWriteTimestamp =
                        haService.getDefaultMessageStore().getSystemClock().now();
                }
            });
        }

        public long getNextTransferFromWhere() {
            return this.nextTransferFromWhere;
        }

        /**
         * 构建给slave的握手消息
         * */
        private boolean buildHandshakeBuffer() {
            final List<EpochEntry> epochEntries = AutoSwitchHAConnection.this.epochCache.getAllEntries();
            final int lastEpoch = AutoSwitchHAConnection.this.epochCache.lastEpoch();
            final long maxPhyOffset = AutoSwitchHAConnection.this.haService.getDefaultMessageStore().getMaxPhyOffset();
            this.byteBufferHeader.position(0);
            this.byteBufferHeader.limit(HANDSHAKE_HEADER_SIZE);
            // State
            this.byteBufferHeader.putInt(currentState.ordinal());
            // Body size
            this.byteBufferHeader.putInt(epochEntries.size() * EPOCH_ENTRY_SIZE);
            // Offset
            this.byteBufferHeader.putLong(maxPhyOffset); // master目前最大偏移量
            // Epoch
            this.byteBufferHeader.putInt(lastEpoch); // master的最新任期
            this.byteBufferHeader.flip();

            // EpochEntries
            this.handShakeBuffer.position(0);
            this.handShakeBuffer.limit(EPOCH_ENTRY_SIZE * epochEntries.size());
            for (final EpochEntry entry : epochEntries) {
                if (entry != null) {
                    this.handShakeBuffer.putInt(entry.getEpoch());
                    this.handShakeBuffer.putLong(entry.getStartOffset());
                }
            }
            this.handShakeBuffer.flip();
            LOGGER.info("Master build handshake header: maxEpoch:{}, maxOffset:{}, epochEntries:{}", lastEpoch, maxPhyOffset, epochEntries);
            return true;
        }

        private boolean handshakeWithSlave() throws IOException {
            // Write Header
            boolean result = this.haWriter.write(this.socketChannel, this.byteBufferHeader);

            if (!result) {
                return false;
            }

            // Write Body
            return this.haWriter.write(this.socketChannel, this.handShakeBuffer);
        }

        // Normal transfer method
        private void buildTransferHeaderBuffer(long nextOffset, int bodySize) {

            EpochEntry entry = AutoSwitchHAConnection.this.epochCache.getEntry(AutoSwitchHAConnection.this.currentTransferEpoch);

            if (entry == null) {

                // If broker is started on empty disk and no message entered (nextOffset = -1 and currentTransferEpoch = -1), do not output error log when sending heartbeat
                if (nextOffset != -1 || currentTransferEpoch != -1 || bodySize > 0) {
                    LOGGER.error("Failed to find epochEntry with epoch {} when build msg header", AutoSwitchHAConnection.this.currentTransferEpoch);
                }

                if (bodySize > 0) {
                    return;
                }
                // Maybe it's used for heartbeat
                entry = AutoSwitchHAConnection.this.epochCache.firstEntry();
            }
            // Build Header
            this.byteBufferHeader.position(0);
            this.byteBufferHeader.limit(TRANSFER_HEADER_SIZE);
            // State
            this.byteBufferHeader.putInt(currentState.ordinal());
            // Body size
            this.byteBufferHeader.putInt(bodySize);
            // Offset
            this.byteBufferHeader.putLong(nextOffset);
            // Epoch
            this.byteBufferHeader.putInt(entry.getEpoch());
            // EpochStartOffset
            this.byteBufferHeader.putLong(entry.getStartOffset());
            // Additional info(confirm offset)
            final long confirmOffset = AutoSwitchHAConnection.this.haService.getDefaultMessageStore().getConfirmOffset();
            this.byteBufferHeader.putLong(confirmOffset);
            this.byteBufferHeader.flip();
        }

        private boolean sendHeartbeatIfNeeded() throws Exception {
            long interval = haService.getDefaultMessageStore().getSystemClock().now() - this.lastWriteTimestamp;
            if (interval > haService.getDefaultMessageStore().getMessageStoreConfig().getHaSendHeartbeatInterval()) {
                buildTransferHeaderBuffer(this.nextTransferFromWhere, 0);
                return this.transferData(0);
            }
            return true;
        }

        private void transferToSlave() throws Exception {
            /**
             * lastWriteOver 之前说过，代表着一次数据传输是否完成
             * 因为 rocketmq的数据传输分两次
             * 1、先传输头部
             * 2、再传body
             * */
            if (this.lastWriteOver) {
                // 如果不是在数据传输过程中，那么尝试发送一次心跳到slave
                this.lastWriteOver = sendHeartbeatIfNeeded();
            } else {
                // maxTransferSize == -1 means to continue transfer remaining data.

                // todo 否则继续传输
                this.lastWriteOver = this.transferData(-1);
            }
            // body数据传输完成
            if (!this.lastWriteOver) {
                return;
            }

            // todo 否则开始组装body数据
            int size = this.getNextTransferDataSize();

            /**
             * 是否有可传输的数据
             * */
            if (size > 0) {

                // 是否支持批次传输，支持的默认可以传32k的数据一次
                if (size > haService.getDefaultMessageStore().getMessageStoreConfig().getHaTransferBatchSize()) {
                    size = haService.getDefaultMessageStore().getMessageStoreConfig().getHaTransferBatchSize();
                }
                int canTransferMaxBytes = flowMonitor.canTransferMaxByteNum();
                if (size > canTransferMaxBytes) {
                    if (System.currentTimeMillis() - lastPrintTimestamp > 1000) {
                        LOGGER.warn("Trigger HA flow control, max transfer speed {}KB/s, current speed: {}KB/s",
                            String.format("%.2f", flowMonitor.maxTransferByteInSecond() / 1024.0),
                            String.format("%.2f", flowMonitor.getTransferredByteInSecond() / 1024.0));
                        lastPrintTimestamp = System.currentTimeMillis();
                    }
                    size = canTransferMaxBytes;
                }

                // 没有可传输的数据大小，注意：不是没有可传输的数据，而是限制允许传输的数据大小 <= 0
                if (size <= 0) {
                    this.releaseData();
                    this.waitForRunning(100);
                    return;
                }

                /**
                 * We must ensure that the transmitted logs are within the same epoch
                 * If currentEpochEndOffset == -1, means that currentTransferEpoch = last epoch, so the endOffset = Long.max
                 *
                 * 更新currentTransferEpochEndOffset
                 * currentTransferEpochEndOffset 是正在传输的epoch的结束消息偏移量，怎么理解呢
                 * 如果现在有两个epoch 1、2
                 * 如果此时传输的消息属于1(如：初始同步时),因为1并不是最新的epoch，所以currentTransferEpochEndOffset会是epoch = 2的第一个消息的起始偏移量
                 * 如果传输消息的epoch是最新的(如：2),那么currentTransferEpochEndOffset固定为-1(因此集群正处于这个epoch，结束偏移量是多少自然不知道)
                 * @see AutoSwitchHAConnection#changeTransferEpochToNext(org.apache.rocketmq.remoting.protocol.EpochEntry)
                 *
                 * 这里的判断本质是为了控制数据的传输在同一个eopch内，具体往下看
                 *
                 * 条件
                 * 1、currentEpochEndOffset != -1 说明就是在传输的不是最新epoch的消息
                 *
                 * 2、否则判断this.nextTransferFromWhere + size > currentEpochEndOffset
                 * 说明这次发送的消息(size)不仅仅只是在epoch1了，因为他都超出了epoch1的结束索引了，因此里面有一部分信息肯定是epoch2的了
                 * 这里就需要控制，不混发
                 * */
                final long currentEpochEndOffset = AutoSwitchHAConnection.this.currentTransferEpochEndOffset;
                if (currentEpochEndOffset != -1 && this.nextTransferFromWhere + size > currentEpochEndOffset) {

                    // 校验epoch，既然currentEpochEndOffset ！= -1，那说明一定是有新的epoch产生了，所以这里校验下新的epoch是否存在(当前传输的epoch + 1)
                    final EpochEntry epochEntry = AutoSwitchHAConnection.this.epochCache.nextEntry(AutoSwitchHAConnection.this.currentTransferEpoch);
                    if (epochEntry == null) {
                        LOGGER.error("Can't find a bigger epochEntry than epoch {}", AutoSwitchHAConnection.this.currentTransferEpoch);
                        waitForRunning(100);
                        return;
                    }
                    /**
                     * 控制发送的消息大小在同一个epoch之内
                     * */
                    size = (int) (currentEpochEndOffset - this.nextTransferFromWhere);
                    /**
                     * 更新下一个新的epoch 和 结束索引
                     * */
                    changeTransferEpochToNext(epochEntry);
                }

                // todo 当前传输数据的起始偏移量
                this.transferOffset = this.nextTransferFromWhere;

                // todo 下次传输数据的起始偏移量
                this.nextTransferFromWhere += size;
                updateLastTransferInfo();

                // Build Header
                buildTransferHeaderBuffer(this.transferOffset, size);  // 构建消息头部

                this.lastWriteOver = this.transferData(size); // 传输数据
            } else {
                /**
                 * 即便是没有数据传输，那么也需要去更新数据传递时间
                 * 以确保不会被剔除同步副本集合
                 * */
                // If size == 0, we should update the lastCatchupTimeMs
                AutoSwitchHAConnection.this.haService.updateConnectionLastCaughtUpTime(AutoSwitchHAConnection.this.slaveId, System.currentTimeMillis());
                haService.getWaitNotifyObject().allWaitForRunning(100);
            }
        }

        @Override
        public void run() {
            AutoSwitchHAConnection.LOGGER.info(this.getServiceName() + " service started");

            while (!this.isStopped()) {
                try {
                    this.selector.select(1000);

                    switch (currentState) {
                        case HANDSHAKE:
                            // Wait until the slave send it handshake msg to master.
                            // todo 等待slave先发握手消息
                            if (!isSlaveSendHandshake) {
                                this.waitForRunning(10);
                                continue;
                            }

                            if (this.lastWriteOver) {
                                if (!buildHandshakeBuffer()) { // todo 构建握手请求
                                    LOGGER.error("AutoSwitchHAConnection build handshake buffer failed");
                                    this.waitForRunning(5000);
                                    continue;
                                }
                            }

                            this.lastWriteOver = handshakeWithSlave();
                            if (this.lastWriteOver) {
                                // change flag to {false} to wait for slave notification
                                isSlaveSendHandshake = false;
                            }
                            break;
                        case TRANSFER:
                            /**
                             * 主从进入传输状态，master可以开始向slave同步数据
                             * */

                            /**
                             * 如果slaveRequestOffset = -1，说明slave还没向master上报截断偏移量
                             * 因为数据传输开始之前，master 和 slave需要通过握手协议来对齐一个传输偏移量
                             * 对齐之后才能还是传输，这个偏移量就存在slaveRequestOffset
                             * */
                            if (-1 == slaveRequestOffset) {
                                this.waitForRunning(10);
                                continue;
                            }

                            /**
                             * 注意：数据传输并不是完全依赖slave上报的偏移量
                             * 而是master自身会维护nextTransferFromWhere,但第一次时依赖slave上报的偏移量进行
                             * -1 == this.nextTransferFromWhere 初次传输
                             * */
                            if (-1 == this.nextTransferFromWhere) {
                                // todo 如果slaveRequestOffset = 0，说明master 和 slave没有对齐的偏移量
                                //      这是master从什么地方开始传输就有下面两个逻辑
                                if (0 == slaveRequestOffset) {
                                    // We must ensure that the starting point of syncing log
                                    // must be the startOffset of a file (maybe the last file, or the minOffset)
                                    final MessageStoreConfig config = haService.getDefaultMessageStore().getMessageStoreConfig();

                                    // slave告诉了master，从最新的日志文件开始传输，那么nextTransferFromWhere 就是最新文件的第一个消息的起始偏移量
                                    if (AutoSwitchHAConnection.this.isSyncFromLastFile) {
                                        long masterOffset = haService.getDefaultMessageStore().getCommitLog().getMaxOffset();
                                        masterOffset = masterOffset - (masterOffset % config.getMappedFileSizeCommitLog());
                                        if (masterOffset < 0) {
                                            masterOffset = 0;
                                        }
                                        this.nextTransferFromWhere = masterOffset;
                                    } else {
                                        // todo 否则就从最小偏移量开始传输(也就是目前master拥有的最小commitLog偏移量)
                                        this.nextTransferFromWhere = haService.getDefaultMessageStore().getCommitLog().getMinOffset();
                                    }
                                } else {
                                    // 否则话slaveRequestOffset 作为第一次传输的偏移量
                                    this.nextTransferFromWhere = slaveRequestOffset;
                                }

                                // nextTransferFromWhere is not found. It may be empty disk and no message is entered
                                /**
                                 * 如果到这里传输偏移量还是为-1，那说明一点，master没有任何数据。
                                 * 这时就无需传输数据，向slave发送一个心跳请求即可
                                 * */
                                if (this.nextTransferFromWhere == -1) {
                                    sendHeartbeatIfNeeded();
                                    waitForRunning(500);
                                    break;
                                }
                                // Setup initial transferEpoch
                                // todo 找到nextTransferFromWhere所对应的epochEntry(这实际就是对应一个消息)
                                EpochEntry epochEntry = AutoSwitchHAConnection.this.epochCache.findEpochEntryByOffset(this.nextTransferFromWhere);
                                // 如果epochEntry 为空，说明有偏移量对应的消息都不存在，这种异常直接返回
                                if (epochEntry == null) {
                                    LOGGER.error("Failed to find an epochEntry to match nextTransferFromWhere {}", this.nextTransferFromWhere);
                                    sendHeartbeatIfNeeded();
                                    waitForRunning(500);
                                    break;
                                }
                                // todo 更新当前传输的epoch，也就是此时数据的传输在哪个epoch中
                                changeTransferEpochToNext(epochEntry);
                                LOGGER.info("Master transfer data to slave {}, from offset:{}, currentEpoch:{}",
                                    AutoSwitchHAConnection.this.clientAddress, this.nextTransferFromWhere, epochEntry);
                            }
                            /**
                             *  传输数据到slave
                             * */
                            transferToSlave();
                            break;
                        default:
                            throw new Exception("unexpected state " + currentState);
                    }
                } catch (Exception e) {
                    AutoSwitchHAConnection.LOGGER.error(this.getServiceName() + " service has exception.", e);
                    break;
                }
            }

            this.onStop();

            changeCurrentState(HAConnectionState.SHUTDOWN);

            this.makeStop();

            readSocketService.makeStop();

            haService.removeConnection(AutoSwitchHAConnection.this);

            SelectionKey sk = this.socketChannel.keyFor(this.selector);
            if (sk != null) {
                sk.cancel();
            }

            try {
                this.selector.close();
                this.socketChannel.close();
            } catch (IOException e) {
                AutoSwitchHAConnection.LOGGER.error("", e);
            }

            flowMonitor.shutdown(true);

            AutoSwitchHAConnection.LOGGER.info(this.getServiceName() + " service end");
        }

        abstract protected int getNextTransferDataSize();

        abstract protected void releaseData();

        abstract protected boolean transferData(int maxTransferSize) throws Exception;

        abstract protected void onStop();
    }
}
